/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.samples.rtp;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Properties;

import javax.media.protocol.ContentDescriptor;
import javax.media.protocol.PushSourceStream;
import javax.media.protocol.SourceTransferHandler;
import javax.media.rtp.OutputDataStream;
import javax.media.rtp.RTPConnector;

import cgl.narada.matching.rtp.RtpEvent;
import cgl.narada.matching.rtp.RtpProfile;
import cgl.narada.protocol.DataReceived;
import cgl.narada.protocol.Handler;
import cgl.narada.protocol.Protocol;
import cgl.narada.transport.TransportException;
import cgl.narada.transport.TransportHandler;
import cgl.narada.transport.TransportHandlerImpl;
import cgl.narada.util.ByteUtilities;

/**
 * An implementation of RTPConnector based on Naradabrokering RTP support.
 *
 * In initialization phase, this might fail.
 * because when we register this client we are sending the message with UDP
 * If that message of the response of that message get lost, then this client will not initialize
 * Therefore it will not work.
 */
public class NaradaRTPConnector implements RTPConnector, Handler {

   private String ip;
   private int remotePort;
   private int localPort;
   private int receiveBufferSize;
   private int sendBufferSize;
   private NaradaRTPInputStream dataInputStream = null;
   private NaradaRTPInputStream controlInputStream = null;
   private NaradaRTPOutputStream dataOutputStream = null;
   private NaradaRTPOutputStream controlOutputStream = null;
   protected TransportHandler transportHandler;
   private int dataTopicNumber;
   private int controlTopicNumber;

   // this integer assigned by the broker to this endPoint
   // to distinguish this client from others
   protected int endPointID;

   // this is a String which shows the broker this client will connect
   // it is in the form of  udp://<ip-address>:portNumber
   protected String brokerURI = null;

   public NaradaRTPConnector(String ip, int remotePort, int topicNumber) {
      this.ip = ip;
      this.remotePort = remotePort;

      dataTopicNumber = topicNumber;
      controlTopicNumber = topicNumber + 1;

      transportHandler = new TransportHandlerImpl(this);

      //load communication channel
      loadUdpCommunicationType();

      //set up broker URI
      setupBrokerURI();

      //wait for a while
      synchronized(this){
         //send the request to Narada Server
         sendEndPointIdRequest();

         try{
            this.wait();
         }catch(Exception e){
            e.printStackTrace();
         }
      }

      //subscribe to the data topic
      subscribeTo(dataTopicNumber);

      //subscribe to the control topic
      subscribeTo(controlTopicNumber);
   }

   /**
    * loads the udp communication type to transport handler
    */
   private void loadUdpCommunicationType() {
      //load communication channel
      Properties properties = new Properties();
      localPort = getRandomPortNumber();
      properties.put("UDPListenerPort", Integer.toString(localPort));

      try { 
         transportHandler.loadCommunicationsOfType(properties, "udp");
      } catch (TransportException transEx) {
         System.out.println(transEx);
      }
   }

   /**
    * sets up the brokerURI for this client
    */
   private void setupBrokerURI() {
      Properties properties = new Properties();
      properties.put("hostname", ip);
      properties.put("portnum", Integer.toString(remotePort));

      try {
         brokerURI = transportHandler.setupLink(properties, "udp");
         System.out.println("Created broker URI " + brokerURI);
      } catch (TransportException transEx) {
         System.out.println(transEx);
      }
   }

   /**
    * send an endPointId request to the broker.
    */
   private void sendEndPointIdRequest() {
      byte[] data = new byte[1];
      data[0] = Protocol.NAP_CLIENT_ADD;
      sendDataTo(data);
   }

   /**
    * this method will be invoked by handleData method, when a response is received for the request
    *
    * @param sender
    * @param addressResponse
    */
   public void processNodeAddressResponse(String sender, byte[] addressResponse) {
      if (addressResponse.length == 1) {
         System.out.println("This server node (" + sender + ") hasn't been assigned an address yet ");
      }

      if (addressResponse.length == 2) {
         System.out.println("The system has exceeded its capacity to handle" + " clients, please try at a different node ");
      }

      int offset = 0;
      offset++;

      byte[] destinationBytes = new byte[4];
      System.arraycopy(addressResponse, offset, destinationBytes, 0, 4);
      endPointID = ByteUtilities.getInt(destinationBytes);
      System.out.println("End point Id= " + endPointID);
      synchronized(this){
         this.notify();
      }
   }

   /**
    * @param dataReceived received data
    */
   public void handleData(byte[] data, String sender) {
      if (data.length == 0) {
         System.out.println(this.getClass().getName() + ": Data received has length = 0");

         return;
      }

      if (data[0] == Protocol.NARADA_RTP_EVENT) {
         RtpEvent rtpEvent = new RtpEvent(data);

         if (rtpEvent.getTopic() == dataTopicNumber) {
            if (dataInputStream != null) {
               dataInputStream.onMessage(rtpEvent);
            } else {
               System.err.println(this.getClass().getName() + ": Data received but dataInputStream is null.");
            }
         } else if (rtpEvent.getTopic() == controlTopicNumber) {
            if (controlInputStream != null) {
               controlInputStream.onMessage(rtpEvent);
            } else {
               //               System.err.println(this.getClass().getName()+": Data received but controlInputStream is null.");
            }
         } else {
            System.err.println("Topic name does not match" + "dataTopicNumber: " + dataTopicNumber + "controlTopicNumber: " + controlTopicNumber + "Received message topic name: " + rtpEvent.getTopic());
         }
      } else if (data[0] == Protocol.NAP_CLIENT_ADD_RESPONSE) {
         processNodeAddressResponse(sender, data);
      } else {
         System.err.println("Error::Data[0] ->" + data[0] + " Data length = " + data.length);
      }
   }

   /**
    * This method indicates that the node in question is a client
    */
   public boolean isClient() {
      return true;
   }

   /**
    * subscribe to a topic on the given node.
    */
   public void subscribeTo(int topic) {
      RtpProfile rtpProfile = new RtpProfile(topic, brokerURI, endPointID);
      byte[] profileBytes = rtpProfile.getBytes();
      int profileLength = profileBytes.length;
      ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
      DataOutputStream dout = new DataOutputStream(baOutputStream);

      try {
         dout.writeByte(Protocol.RTP_CLIENT_ADD_PROFILE);
         dout.writeInt(profileLength);
         dout.write(profileBytes);
      } catch (IOException ioe) {
         System.out.println(this.getClass().getName() + ": Error marshalling stream" + ioe);
      }/* end try-catch */

      sendDataTo(baOutputStream.toByteArray());
   }

   /**
    * send the given byte array
    */
   public void sendDataTo(byte[] data) {
      try {
         transportHandler.sendData(data, brokerURI);
      } catch (TransportException transEx) {
         System.out.println(transEx);
      }
   }

   /**
    * coverts the received data to a serialized RtpEvent object and passes it to the transportHandler
    */
   public void propagateEventToBroker(int topic, byte[] data, int offset, int length) {
      RtpEvent rtpEvent = new RtpEvent(endPointID, topic, data, offset, length);
      byte[] rtpEventBytes = rtpEvent.getBytesWithClientFlag();
      sendDataTo(rtpEventBytes);
   }

   /**
    * returns an even random port number in the range of 10,000 - 50,000
    * @return
    */
   private int getRandomPortNumber() {

      int port;
      boolean tryAnother = true;

      do {
         port = (int)(Math.random() * 40000) + 10000;

         try {
            java.net.DatagramSocket ds = new java.net.DatagramSocket(port);
            ds.close();
            tryAnother = false;
         } catch (java.net.SocketException se) {
            tryAnother = true;
         }
      } while (tryAnother);

      return port;
   }

   /**
    * return the endPointId
    */
   public int getEndPointId() {
      return this.endPointID;
   }

   /**
   * Returns an input stream to receive the RTP data.
   */
   public PushSourceStream getDataInputStream() {
      if (dataInputStream == null) {
         dataInputStream = new NaradaRTPInputStream(dataTopicNumber);
      }

      return dataInputStream;
   }

   /**
   * Returns an output stream to send the RTP data.
   */
   public OutputDataStream getDataOutputStream() {
      if (dataOutputStream == null) {
         dataOutputStream = new NaradaRTPOutputStream(dataTopicNumber);
      }

      return dataOutputStream;
   }

   /**
   * Returns an input stream to receive the RTCP data.
   */
   public PushSourceStream getControlInputStream() {
      if (controlInputStream == null) {
         controlInputStream = new NaradaRTPInputStream(controlTopicNumber);
      }

      return controlInputStream;
   }

   /**
   * Returns an output stream to send the RTCP data.
   */
   public OutputDataStream getControlOutputStream() {
      if (controlOutputStream == null) {
         controlOutputStream = new NaradaRTPOutputStream(controlTopicNumber);
      }

      return controlOutputStream;
   }

   /**
   * Close all the RTP, RTCP streams.
   */
   public void close() {
      transportHandler.closeAllLinks();
      transportHandler.disposeAllCommunicationResources();
   }

   /**
   * Set the receive buffer size of the RTP data channel.
   * This is only a hint to the implementation.  The actual implementation
   * may not be able to do anything to this.
   */
   public void setReceiveBufferSize(int size) throws IOException {
      receiveBufferSize = size;
   }

   /**
   * Get the receive buffer size set on the RTP data channel.
   * Return -1 if the receive buffer size is not applicable for
   * the implementation.
   */
   public int getReceiveBufferSize() {
      return receiveBufferSize;
   }

   /**
   * Set the send buffer size of the RTP data channel.
   * This is only a hint to the implementation.  The actual implementation
   * may not be able to do anything to this.
   */
   public void setSendBufferSize(int size) throws IOException {
      sendBufferSize = size;
   }

   /**
   * Get the send buffer size set on the RTP data channel.
   * Return -1 if the send buffer size is not applicable for
   * the implementation.
   */
   public int getSendBufferSize() {
      return sendBufferSize;
   }

   /**
   * Return the RTCP bandwidth fraction.  This value is used to
   * initialize the RTPManager.  Check RTPManager for more detauls.
   * Return -1 to use the default values.
   */
   public double getRTCPBandwidthFraction() {
      return -1;
   }

   /**
   * Return the RTCP sender bandwidth fraction.  This value is used to
   * initialize the RTPManager.  Check RTPManager for more detauls.
   * Return -1 to use the default values.
   */
   public double getRTCPSenderBandwidthFraction() {
      return -1;
   }

   /**
   * An inner class to implement an OutputDataStream based on UDP sockets.
   */
   class NaradaRTPOutputStream implements OutputDataStream {

      private int topicNumber;
      boolean isDataStream;

      public NaradaRTPOutputStream(int topicNumber) {
         this.topicNumber = topicNumber;

         if (topicNumber == dataTopicNumber) {
            isDataStream = true;
         } else {
            isDataStream = false;
         }

      }

      public int write(byte[] data, int offset, int len) {

         //send the package
         try {
            propagateEventToBroker(topicNumber, data, offset, len);
         } catch (Exception e) {
            e.printStackTrace();

            return -1;
         }

         return len;
      }

   } // end of class NaradaRTPOutputStream

   /**
   * An inner class to implement an PushSourceStream based on UDP sockets.
   */
   class NaradaRTPInputStream extends Thread implements PushSourceStream {

      SourceTransferHandler sth = null;
      RtpEvent message;
      int topicNumber;
      Object synchObject = new Object();

      public NaradaRTPInputStream(int topicNumber) {
         this.topicNumber = topicNumber;
      }

      public int read(byte[] buffer, int offset, int length) {
         synchronized (synchObject) {
            System.arraycopy(message.getBuffer(), message.getOffset(), buffer, offset, message.getDataLength());
         }

         return message.getDataLength();
      }

      public void onMessage(RtpEvent message) {
         synchronized (synchObject) {
            this.message = message;
         }

         if (sth != null) {
            sth.transferData(this);
         } else {
            System.out.println(this.getClass().getName() + ": sth is null .......................");
         }
      }

      public int getMinimumTransferSize() {
         return 2 * 1024; // twice the MTU size, just to be safe.
      }

      public synchronized void setTransferHandler(SourceTransferHandler sth) {
         this.sth = sth;
      }

      // Not applicable.
      public ContentDescriptor getContentDescriptor() {
         return null;
      }

      // Not applicable.
      public long getContentLength() {
         return LENGTH_UNKNOWN;
      }

      // Not applicable.
      public boolean endOfStream() {
         return false;
      }

      // Not applicable.
      public Object[] getControls() {
         return new Object[0];
      }

      // Not applicable.
      public Object getControl(String type) {
         return null;
      }
   } // end of class NaradaRTPInputStream
}
